package com.zengjianjun.hadoop.demo.client;

import com.zengjianjun.hadoop.demo.config.props.HadoopProperties;
import com.zengjianjun.hadoop.demo.util.FileUtil;
import lombok.AllArgsConstructor;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;
import org.springframework.web.multipart.MultipartFile;

import javax.annotation.PostConstruct;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

/**
 * @author zengjianjun
 */
@Slf4j
@Component
@AllArgsConstructor
public class HadoopClient {

    private FileSystem fileSystem;

    private HadoopProperties hadoopProperties;


    @PostConstruct
    public void init() {
        log.info("HadoopClient init start...");
        boolean mkdir = this.mkdir(hadoopProperties.getDirectoryPath());
        log.info("HadoopClient init end... {}", mkdir);
    }

    /**
     * 创建目录
     *
     * @param folderPath 文件夹路径名称
     * @return {@link Boolean}
     */
    public boolean mkdir(String folderPath) {
        log.info("[开始创建目录] 文件夹路径名称: {}", folderPath);
        if (StringUtils.isEmpty(folderPath)) {
            throw new IllegalArgumentException("folder不能为空");
        }
        try {
            Path path = new Path(folderPath);
            if (!fileSystem.exists(path)) {
                return fileSystem.mkdirs(path);
            }
            return true;
        } catch (Exception e) {
            log.error("[创建目录失败]", e);
            return false;
        }
    }

    /**
     * 文件上传
     *
     * @param file 文件信息
     */
    public void upload(MultipartFile file) {
        this.upload(FileUtil.transferToFile(file));
    }

    /**
     * 文件上传
     *
     * @param file 文件信息
     */
    public void upload(File file) {
        this.upload(file, file.getName());
    }

    /**
     * 文件上传
     *
     * @param file     文件信息
     * @param fileName 文件名称
     */
    public void upload(File file, String fileName) {
        this.upload(true, true, file.getPath(), StringUtils.isNotEmpty(fileName) ? fileName : file.getName());
    }

    /**
     * 文件上传
     *
     * @param delSrc      是否删除源文件，true为删除，默认为false
     * @param overwrite   是否覆盖
     * @param srcFile     源文件，上传文件路径
     * @param desFileName fs的目标路径
     */
    public void upload(boolean delSrc, boolean overwrite, String srcFile, String desFileName) {
        log.info("[文件上传] 开始上传, 上传文件名称: {}", desFileName);
        Path srcPath = new Path(srcFile);
        // 目标路径
        Path dstPath = new Path(hadoopProperties.getFullFilePath(desFileName));
        try {
            // 文件上传
            fileSystem.copyFromLocalFile(delSrc, overwrite, srcPath, dstPath);
        } catch (IOException e) {
            log.error("[文件上传失败]", e);
        }
    }

    /**
     * 下载文件
     *
     * @param fileName     文件名称
     * @param outputStream 输出流
     */
    public void download(@NotNull String fileName, OutputStream outputStream) {
        try {
            log.info("[下载文件] 开始下载, 下载文件名称: {}", fileName);
            @Cleanup InputStream is = fileSystem.open(new Path(hadoopProperties.getFullFilePath(fileName)));
            IOUtils.copyBytes(is, outputStream, 4096, true);
        } catch (Exception e) {
            log.error("[下载文件失败]", e);
        }

    }

    /**
     * 删除文件或者文件目录
     *
     * @param fullPath 文件目录完整路径
     */
    public void rmdir(@NotNull String fullPath) {
        log.info("[删除文件] 开始删除, 删除文件目录的路径: {}", fullPath);
        try {
            Path path = new Path(fullPath);
            if (fileSystem.exists(path)) {
                fileSystem.delete(path, true);
            }
        } catch (Exception e) {
            log.error("[删除文件失败]", e);
        }
    }

    /**
     * 下载文件到本地
     *
     * @param fileName     文件路径
     * @param downloadPath 本地下载路径
     */
    public void downloadFileFromLocal(String fileName, String downloadPath) {
        log.info("[下载文件到本地] 开始下载, 文件名称: {}, 本地下载路径: {}", fileName, downloadPath);
        // 上传路径
        Path clientPath = new Path(hadoopProperties.getFullFilePath(fileName));
        // 目标路径
        Path serverPath = new Path(downloadPath);
        try {
            // 调用文件系统的文件复制方法，第一个参数是否删除原文件true为删除，默认为false
            fileSystem.copyToLocalFile(false, clientPath, serverPath);
        } catch (IOException e) {
            log.info("[下载文件到本地失败]", e);
        }
    }
}
